package com.jacky.sender;

import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.MediaType;
import org.springframework.http.RequestEntity;
import org.springframework.web.client.RestTemplate;
import zipkin2.Call;
import zipkin2.Callback;
import zipkin2.Span;
import zipkin2.codec.BytesEncoder;
import zipkin2.codec.Encoding;
import zipkin2.reporter.BytesMessageEncoder;
import zipkin2.reporter.Sender;
import zipkin2.reporter.internal.BaseCall;

import java.io.IOException;
import java.net.URI;
import java.util.List;

import static zipkin2.codec.SpanBytesEncoder.JSON_V2;

/**
 * Created with IntelliJ IDEA.
 *
 * @author: jacky
 * @date: 2018/7/31
 * Time: 9:25
 * Description:
 **/
public class MyRestTemplateSender extends Sender {

    RestTemplate restTemplate;
    String url;
    Encoding encoding;
    MediaType mediaType;
    BytesMessageEncoder messageEncoder;

    transient boolean closeCalled;

    MyRestTemplateSender(RestTemplate restTemplate, String baseUrl, BytesEncoder<Span> encoder) {
        this.restTemplate = restTemplate;
        this.encoding = encoder.encoding();
        if (encoder.equals(JSON_V2)) {
            this.mediaType = MediaType.APPLICATION_JSON;
            this.url = baseUrl + (baseUrl.endsWith("/") ? "" : "/") + "api/v2/spans";
        } else if (this.encoding == Encoding.JSON) {
            this.mediaType = MediaType.APPLICATION_JSON;
            this.url = baseUrl + (baseUrl.endsWith("/") ? "" : "/") + "api/v1/spans";
        } else {
            throw new UnsupportedOperationException("Unsupported encoding: " + this.encoding.name());
        }
       // this.url="http://127.0.0.1:8100/api/v1/spans";
        this.messageEncoder = BytesMessageEncoder.forEncoding(this.encoding);
    }

    public void setRestTemplate(RestTemplate restTemplate) {
        this.restTemplate = restTemplate;
    }

    @Override
    public Encoding encoding() {
        return this.encoding;
    }

    @Override
    public int messageMaxBytes() {
        return 5 * 1024 * 1024;
    }

    @Override
    public int messageSizeInBytes(List<byte[]> spans) {
        return encoding().listSizeInBytes(spans);
    }


    /**
     * Sends a list of encoded spans to a transport such as http or Kafka.
     *
     * @param encodedSpans
     * @return
     */
    @Override
    public Call<Void> sendSpans(List<byte[]> encodedSpans) {
        if(this.closeCalled){
            throw new IllegalStateException("close");
        }
        for (byte[] span: encodedSpans) {
            String strRead = new String(span);
            strRead=String.copyValueOf(strRead.toCharArray(), 0, span.length);

        }
        return new HttpPostCall(this.messageEncoder.encode(encodedSpans));
    }


    class HttpPostCall extends BaseCall<Void> {
        private final byte[] message;

        HttpPostCall(byte[] message) {
            this.message = message;
        }

        @Override
        protected Void doExecute() throws IOException {
            post(this.message);
            return null;
        }

        @Override
        protected void doEnqueue(Callback<Void> callback) {
            try {
                post(this.message);
                callback.onSuccess(null);
            } catch (Exception e) {
                callback.onError(e);
            }
        }

        @Override
        public Call<Void> clone() {
            return new HttpPostCall(this.message);
        }
    }

    void post(byte[] json) {
        HttpHeaders httpHeaders = new HttpHeaders();
        httpHeaders.setContentType(this.mediaType);
        RequestEntity<byte[]> requestEntity = new RequestEntity<byte[]>(json, httpHeaders,
                HttpMethod.POST, URI.create(this.url));
        this.restTemplate.exchange(requestEntity, String.class);
    }
}